//package com.mq.kafkamain;
//
//import org.apache.commons.lang3.StringUtils;
//import org.apache.kafka.clients.consumer.ConsumerRecord;
//import org.apache.kafka.clients.consumer.ConsumerRecords;
//import org.apache.kafka.clients.consumer.KafkaConsumer;
//
//import java.time.Duration;
//import java.util.Arrays;
//import java.util.Properties;
//
///**
// * kafka消费者帮助类
// * <p>
// * 可以用spring bean方式
// *
// * @author cl
// */
//public class KafkaConsumerBuilder {
//    /**
//     * kafka地址
//     */
//    private String[] bootstrapServers = new String[]{"localhost:9092"};
//
//    private String group;
//
//    public com.sunsea.util.kafka.KafkaConsumerBuilder host(String... bootstrapServers) {
//        this.bootstrapServers = bootstrapServers;
//        return this;
//    }
//
//    public com.sunsea.util.kafka.KafkaConsumerBuilder group(String group) {
//        this.group = group;
//        return this;
//    }
//
//    public KafkaConsumer<String, String> create() {
//        if (this.group == null) {
//            throw new RuntimeException("kafka必须配置所属组");
//        }
//        Properties props = new Properties();
//        props.put("bootstrap.servers", StringUtils.join(bootstrapServers, ","));
//        props.put("group.id", this.group);
//        props.put("enable.auto.commit", "false");
//        props.put("auto.offset.reset", "earliest");
//        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
//        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
//        return new KafkaConsumer<String, String>(props);
//    }
//
//    private static final KafkaConsumer<String, String> CONSUMER = new com.sunsea.util.kafka.KafkaConsumerBuilder()
//            .host("1.15.148.141:9092")
//            .group("rax_group_1")
//            .create();
//
//    public static void main(String[] args) {
//        CONSUMER.subscribe(Arrays.asList("test-topic"));
//        try {
//            while (true) {
//                ConsumerRecords<String, String> records = CONSUMER.poll(Duration.ofMinutes(1));
//                for (ConsumerRecord<String, String> record : records) {
//                    System.err.println(record.offset() + ": " + record.value());
//                }
//                CONSUMER.commitSync();
//            }
//        } finally {
//            CONSUMER.close();
//        }
//
//    }
//
//
//}
